package org.example.flink;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;

/**
 * Created by McClone on 2020/5/16.
 */
public class ParquetSQLExample {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);

        String sql = "CREATE TABLE person (\n" +
                "  name STRING,\n" +
                "  type STRING\n" +
                ") WITH (\n" +
                "  'connector.type' = 'filesystem',\n" +
                "  'connector.path' = 'hdfs://localhost:9000/data/person.parquet/type=1/',\n" +
                "  'format.type' = 'parquet'\n" +
                ")";
        tEnv.sqlUpdate(sql);
        Table result = tEnv.sqlQuery("select * from person");
        tEnv.toAppendStream(result, Person.class).print();
        env.execute("parquetSQL");
    }
}
